AI项目实战（8）AI 视频生成 Agent

class VideoPipelineState:
    def update(self, step: str, status: str, output: str = ""):
        self.steps[step] = {"status": status, "output": output, "timestamp": ...}

    def summary(self) -> str:
        # 返回 ✅🔄❌ 图标化的进度摘要

"""P05 AI 视频生成 Agent - 领域模型层"""
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class PipelineStatus(str, Enum):
    """流水线步骤状态。"""
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class StepResult:
    """单个步骤的执行结果。"""
    step_name: str
    status: PipelineStatus
    output: str = ""
    error_message: str = ""
    start_time: str = field(default_factory=lambda: datetime.now().isoformat())
    end_time: str = ""
    duration_seconds: float = 0.0
    metadata: dict[str, Any] = field(default_factory=dict)

    @property
    def is_success(self) -> bool:
        return self.status == PipelineStatus.SUCCESS


@dataclass
class VideoScript:
    """视频脚本。"""
    topic: str
    duration_minutes: int
    scenes: list[dict[str, Any]] = field(default_factory=list)
    voiceover_text: str = ""
    style: str = "professional"


@dataclass
class Storyboard:
    """分镜表。"""
    script_id: str = ""
    scenes: list[dict[str, Any]] = field(default_factory=list)

    def add_scene(self, scene_number: int, prompt: str, duration: int = 5) -> None:
        self.scenes.append({
            "scene_number": scene_number,
            "prompt": prompt,
            "duration_seconds": duration,
        })


@dataclass
class PipelineState:
    """视频制作流水线状态。"""
    topic: str = ""
    steps: dict[str, StepResult] = field(default_factory=dict)
    final_video_path: str = ""

    def update_step(self, step: str, status: PipelineStatus,
                    output: str = "", error: str = "") -> None:
        """更新步骤状态。"""
        if step not in self.steps:
            self.steps[step] = StepResult(step_name=step, status=status)
        self.steps[step].status = status
        self.steps[step].output = output
        self.steps[step].error_message = error

    def mark_step_end(self, step: str) -> None:
        """标记步骤结束，计算耗时。"""
        if step in self.steps:
            self.steps[step].end_time = datetime.now().isoformat()
            start = datetime.fromisoformat(self.steps[step].start_time)
            end = datetime.fromisoformat(self.steps[step].end_time)
            self.steps[step].duration_seconds = (end - start).total_seconds()

    def get_progress_percent(self) -> int:
        """获取总体进度百分比。"""
        if not self.steps:
            return 0
        completed = sum(1 for s in self.steps.values()
                       if s.status in (PipelineStatus.SUCCESS, PipelineStatus.SKIPPED))
        return int(completed / len(self.steps) * 100)

    def summary_markdown(self) -> str:
        """生成状态摘要（Markdown 格式）。"""
        status_emojis = {
            PipelineStatus.SUCCESS: "✅",
            PipelineStatus.RUNNING: "🔄",
            PipelineStatus.FAILED: "❌",
            PipelineStatus.SKIPPED: "⏭️",
        }
        step_names = {
            "script": "脚本创作",
            "storyboard": "分镜生成",
            "voiceover": "配音生成",
            "video_compose": "视频合成",
        }
        lines = ["## 🎬 视频制作流水线状态", "",
                f"**主题**: {self.topic or '未设置'}",
                f"**进度**: {self.get_progress_percent()}%", "",
                "### 各步骤状态:", ""]
        for step_key, name in step_names.items():
            step = self.steps.get(step_key)
            if step:
                emoji = status_emojis.get(step.status, "⏳")
                lines.append(f"{emoji} **{name}** ({step.duration_seconds:.1f}s)")
                if step.error_message:
                    lines.append(f"   错误: {step.error_message}")
            else:
                lines.append(f"⏳ **{name}** (未开始)")
        return "\n".join(lines)

"""P05 AI 视频生成 Agent - Prompt 层"""
from __future__ import annotations

SCRIPT_PROMPT_TEMPLATE = """请为一段 {duration_minutes} 分钟的短视频创作详细脚本。

**主题**: {topic}

## 输出格式要求

请输出 Markdown 格式，严格遵守以下结构：

# 视频脚本

## 分场景脚本

### 场景 1
**画面描述 (English)**: A visually striking opening shot...

**旁白文本 (中文)**: 大家好，今天我们要聊的是...

**建议时长**: 15 秒

...

## 完整旁白（配音用）
（将所有场景的旁白合并成一段连续的文本）

## 创作要求
1. **场景数量**: {scenes_count} 个场景（每个 10-20 秒）
2. **画面描述**: 用英文，适合作为 AI 图片生成的 prompt，包含景别、光线
3. **旁白文本**: 用中文，口语化、自然，适合朗读
4. **画面描述技巧**: 要包含主体描述、景别、光线、画质、风格
"""


def build_script_prompt(topic: str, duration_minutes: int) -> str:
    """构建脚本创作提示词。"""
    scenes_count = max(5, int(duration_minutes * 4))
    return SCRIPT_PROMPT_TEMPLATE.format(
        topic=topic,
        duration_minutes=duration_minutes,
        scenes_count=scenes_count,
    )


STORYBOARD_PROMPT_TEMPLATE = """请从以下视频脚本中提取每个场景的画面描述，转换为 AI 图片生成 prompt。

## 脚本内容

{script_text}

## 输出要求

请输出 **纯 JSON 数组**，不要其他文字，不要 markdown 标记。格式如下：

[
  {{
    "scene_number": 1,
    "prompt": "Wide shot of modern office, cinematic lighting, 4K, professional",
    "duration_seconds": 15
  }}
]

每个 prompt 必须是英文，包含：景别、光线、画质、风格。
"""


def build_storyboard_prompt(script_text: str) -> str:
    """构建分镜提取提示词。"""
    return STORYBOARD_PROMPT_TEMPLATE.format(script_text=script_text[:4000])

"""P05 AI 视频生成 Agent - 工具层"""
from __future__ import annotations

import json
import re
import time
from pathlib import Path

from langchain.tools import tool

from core.config import DATA_DIR
from core.logging_conf import get_logger

from .models import PipelineState, PipelineStatus
from .prompts import build_script_prompt, build_storyboard_prompt

logger = get_logger("p05.video_generation.tools")
VIDEO_OUTPUT_DIR = DATA_DIR / "videos"
VIDEO_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

# 全局流水线状态
_pipeline = PipelineState()


def _call_llm_for_script(prompt: str) -> str:
    """调用 LLM 生成脚本。生产环境替换为真实的 Claude API 调用。"""
    try:
        from anthropic import Anthropic
        client = Anthropic()
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}],
        )
        return resp.content[0].text
    except ImportError:
        # 开发环境模拟返回
        return "# 视频脚本\n\n## 场景 1\n**画面描述**: Test scene\n**旁白**: 测试旁白"


@tool
def create_script(topic: str, duration_minutes: int = 3) -> str:
    """根据主题创作视频脚本。"""
    logger.info("开始创建脚本: topic='%s'", topic)
    _pipeline.topic = topic
    _pipeline.update_step("script", PipelineStatus.RUNNING)
    
    try:
        prompt = build_script_prompt(topic, duration_minutes)
        script = _call_llm_for_script(prompt)
        
        _pipeline.update_step("script", PipelineStatus.SUCCESS,
                             output=f"脚本已生成，长度 {len(script)} 字符")
        _pipeline.mark_step_end("script")
        return script
    except Exception as e:
        _pipeline.update_step("script", PipelineStatus.FAILED, error=str(e))
        _pipeline.mark_step_end("script")
        return f"脚本创作失败: {e}"


@tool
def generate_storyboard(script: str) -> str:
    """根据脚本生成分镜表。"""
    logger.info("开始生成分镜表")
    _pipeline.update_step("storyboard", PipelineStatus.RUNNING)
    
    try:
        prompt = build_storyboard_prompt(script)
        storyboard_json = _call_llm_for_script(prompt)  # 复用 LLM 调用函数
        
        # 提取并解析 JSON
        json_match = re.search(r'\[.*\]', storyboard_json, re.DOTALL)
        if json_match:
            scenes = json.loads(json_match.group(0))
            _pipeline.update_step("storyboard", PipelineStatus.SUCCESS,
                                 output=f"分镜表已生成，共 {len(scenes)} 个场景")
        else:
            _pipeline.update_step("storyboard", PipelineStatus.SUCCESS,
                                 output="分镜表已生成（JSON 解析未确认）")
        _pipeline.mark_step_end("storyboard")
        return storyboard_json
    except Exception as e:
        _pipeline.update_step("storyboard", PipelineStatus.FAILED, error=str(e))
        _pipeline.mark_step_end("storyboard")
        return f"分镜生成失败: {e}"


@tool
def generate_voiceover(script_text: str, output_filename: str = "voiceover.mp3") -> str:
    """为脚本生成配音。生产环境替换为真实 TTS API。"""
    _pipeline.update_step("voiceover", PipelineStatus.RUNNING)
    try:
        output_path = VIDEO_OUTPUT_DIR / output_filename
        # 模拟实现：写入占位文件
        output_path.write_text(f"[配音占位] 脚本长度: {len(script_text)}", encoding="utf-8")
        _pipeline.update_step("voiceover", PipelineStatus.SUCCESS, output=str(output_path))
        _pipeline.mark_step_end("voiceover")
        return f"配音已生成并保存到: {output_path}"
    except Exception as e:
        _pipeline.update_step("voiceover", PipelineStatus.FAILED, error=str(e))
        return f"配音生成失败: {e}"


@tool
def compose_video(scenes_json: str, output_filename: str = "final_video.mp4") -> str:
    """将各场景画面和配音合成为最终视频。生产环境替换为 FFmpeg。"""
    _pipeline.update_step("video_compose", PipelineStatus.RUNNING)
    try:
        output_path = VIDEO_OUTPUT_DIR / output_filename
        output_path.write_text(f"[视频占位] 场景数: {scenes_json.count('scene')}", encoding="utf-8")
        _pipeline.final_video_path = str(output_path)
        _pipeline.update_step("video_compose", PipelineStatus.SUCCESS, output=str(output_path))
        _pipeline.mark_step_end("video_compose")
        return f"视频已合成并保存到: {output_path}"
    except Exception as e:
        _pipeline.update_step("video_compose", PipelineStatus.FAILED, error=str(e))
        return f"视频合成失败: {e}"


@tool
def get_pipeline_status() -> str:
    """查看当前视频制作流水线的进度状态。"""
    return _pipeline.summary_markdown()

"""P05 AI 视频生成 Agent - 服务层"""
from __future__ import annotations

import time
from typing import Callable

from core.logging_conf import get_logger

from .models import PipelineState, PipelineStatus

logger = get_logger("p05.video_generation.service")


class VideoProductionService:
    """视频制作服务。编排流水线执行、重试、降级。"""

    def __init__(self, max_retries: int = 2, skip_on_failure: bool = True):
        self.max_retries = max_retries
        self.skip_on_failure = skip_on_failure
        self.pipeline = PipelineState()

    def _execute_step(self, step_name: str, func: Callable, *args, **kwargs) -> any:
        """执行单个步骤，包含重试逻辑。"""
        for attempt in range(self.max_retries + 1):
            try:
                logger.info("执行步骤 '%s'，第 %d/%d 次尝试",
                           step_name, attempt + 1, self.max_retries + 1)
                result = func(*args, **kwargs)
                
                if isinstance(result, str) and "失败" in result:
                    raise Exception(result)
                
                logger.info("步骤 '%s' 执行成功", step_name)
                return result
            except Exception as e:
                logger.warning("步骤 '%s' 第 %d 次尝试失败: %s", step_name, attempt + 1, e)
                if attempt < self.max_retries:
                    wait_seconds = 2 ** attempt  # 指数退避
                    logger.info("等待 %d 秒后重试...", wait_seconds)
                    time.sleep(wait_seconds)
                elif self.skip_on_failure:
                    logger.warning("跳过步骤 '%s'，继续后续流程", step_name)
                    return None
                else:
                    raise
        return None

    def create_full_video(self, topic: str, duration_minutes: int = 2) -> PipelineState:
        """执行完整的视频制作流水线。"""
        from .tools import (create_script, generate_storyboard,
                           generate_voiceover, compose_video)
        
        logger.info("开始完整视频制作: topic='%s'", topic)
        self.pipeline.topic = topic

        # 步骤 1: 创建脚本
        script = self._execute_step("script", create_script.invoke,
                                   {"topic": topic, "duration_minutes": duration_minutes})
        # 步骤 2: 生成分镜
        storyboard = None
        if script is not None:
            storyboard = self._execute_step("storyboard", generate_storyboard.invoke,
                                           {"script": script})
        # 步骤 3: 生成配音
        if script is not None:
            self._execute_step("voiceover", generate_voiceover.invoke,
                              {"script_text": script[:2000]})
        # 步骤 4: 合成视频
        if storyboard is not None:
            self._execute_step("video_compose", compose_video.invoke,
                              {"scenes_json": storyboard})

        logger.info("视频制作完成，进度: %d%%", self.pipeline.get_progress_percent())
        return self.pipeline

    def regenerate_step(self, step_name: str, **kwargs) -> any:
        """重新执行某个步骤。"""
        from .tools import (create_script, generate_storyboard,
                           generate_voiceover, compose_video)
        step_funcs = {
            "script": create_script.invoke,
            "storyboard": generate_storyboard.invoke,
            "voiceover": generate_voiceover.invoke,
            "video_compose": compose_video.invoke,
        }
        return self._execute_step(step_name, step_funcs[step_name], kwargs)

"""P05 AI 视频生成 Agent - 项目入口层"""
from __future__ import annotations

from langchain.agents import create_agent

from core import BaseProject, build_chat_model, registry
from core.logging_conf import get_logger

from .prompts import build_system_prompt
from .tools import get_available_tools

logger = get_logger("p05.video_generation.project")


class VideoGenerationProject(BaseProject):
    """AI 视频生成 Agent 项目。"""

    id = "p05_video_generation"
    name = "AI 视频生成 Agent"
    description = "多模态工具编排：脚本→分镜→配音→视频合成"
    capabilities = ["多模态", "流水线", "LLM 编排"]

    def build_agent(self) -> any:
        """构建视频生成 Agent。"""
        model = build_chat_model()
        tools = get_available_tools()
        system_prompt = build_system_prompt()
        return create_agent(model=model, tools=tools, system_prompt=system_prompt)


registry.register(VideoGenerationProject())